use std::any::Any;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use protoactor_rs::actor::{Actor, ActorContext, ActorError, DefaultActorContext, PID, Props};
use protoactor_rs::persistence::{PersistenceError, PersistentEvent, PersistentState};
use protoactor_rs::system::ActorSystem;

// Demo for persistence concept (in current implementation state)
// Since the full integration is still in progress,
// this example just shows the core concepts without using the persistence API directly

// Messages for our counter actor
enum CounterMessage {
    Increment(i32),
    Decrement(i32),
    Get,
    Reset,
}

// Events for persistence
#[derive(Debug, Clone, Serialize, Deserialize)]
enum CounterEvent {
    Incremented(i32),
    Decremented(i32),
    Reset,
}

impl PersistentEvent for CounterEvent {
    fn event_type(&self) -> &str {
        match self {
            CounterEvent::Incremented(_) => "incremented",
            CounterEvent::Decremented(_) => "decremented",
            CounterEvent::Reset => "reset",
        }
    }
}

// State for snapshots
#[derive(Debug, Clone, Serialize, Deserialize)]
struct CounterState {
    count: i32,
    version: u64,
}

impl PersistentState for CounterState {
    fn version(&self) -> u64 {
        self.version
    }
    
    fn set_version(&mut self, version: u64) {
        self.version = version;
    }
}

// Our counter actor with persistence capabilities
struct CounterActor {
    count: i32,
    persist_events: bool,
    events: Vec<CounterEvent>,
}

impl CounterActor {
    fn new(persist_events: bool) -> Self {
        Self { 
            count: 0,
            persist_events,
            events: Vec::new(),
        }
    }
    
    // Apply an event to the actor state
    fn apply_event(&mut self, event: &CounterEvent) {
        match event {
            CounterEvent::Incremented(value) => self.count += value,
            CounterEvent::Decremented(value) => self.count -= value,
            CounterEvent::Reset => self.count = 0,
        }
        
        if self.persist_events {
            self.events.push(event.clone());
        }
    }
    
    // Create a snapshot of the current state
    fn create_snapshot(&self) -> CounterState {
        CounterState {
            count: self.count,
            version: self.events.len() as u64,
        }
    }
    
    // Restore from a snapshot
    fn restore_from_snapshot(&mut self, snapshot: CounterState) {
        self.count = snapshot.count;
        // In a real scenario, we would also track the version to know which events to apply
    }
}

#[async_trait]
impl Actor for CounterActor {
    type Context = DefaultActorContext;
    
    async fn receive(&self, ctx: &mut Self::Context, msg: Box<dyn Any + Send>) -> Result<(), ActorError> {
        let mut this = self.clone();
        
        if let Some(msg) = msg.downcast_ref::<CounterMessage>() {
            match msg {
                CounterMessage::Increment(value) => {
                    // Create and "persist" the event
                    let event = CounterEvent::Incremented(*value);
                    
                    // Apply the event to update state
                    this.apply_event(&event);
                    
                    println!("Counter incremented by {}. New value: {}", value, this.count);
                }
                CounterMessage::Decrement(value) => {
                    let event = CounterEvent::Decremented(*value);
                    this.apply_event(&event);
                    println!("Counter decremented by {}. New value: {}", value, this.count);
                }
                CounterMessage::Reset => {
                    let event = CounterEvent::Reset;
                    this.apply_event(&event);
                    println!("Counter reset to 0");
                }
                CounterMessage::Get => {
                    println!("Current counter value: {}", this.count);
                    ctx.respond(this.count);
                    
                    // Periodic snapshot
                    if this.persist_events && this.events.len() % 5 == 0 && !this.events.is_empty() {
                        let snapshot = this.create_snapshot();
                        println!("Creating snapshot at version {}: {:?}", snapshot.version(), snapshot);
                    }
                }
            }
        }
        
        Ok(())
    }
}

impl Clone for CounterActor {
    fn clone(&self) -> Self {
        Self {
            count: self.count,
            persist_events: self.persist_events,
            events: self.events.clone(),
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize the actor system
    let system = ActorSystem::new()?;
    
    // Create a counter actor with event persistence enabled
    let counter = CounterActor::new(true);
    
    // Create props and spawn the actor
    let props = Props::from_producer(move || counter.clone());
    let counter_pid = system.spawn::<CounterActor>(props).await?;
    
    // Send some messages to the counter
    system.send(&counter_pid, CounterMessage::Get).await?;
    system.send(&counter_pid, CounterMessage::Increment(5)).await?;
    system.send(&counter_pid, CounterMessage::Get).await?;
    system.send(&counter_pid, CounterMessage::Decrement(2)).await?;
    system.send(&counter_pid, CounterMessage::Get).await?;
    system.send(&counter_pid, CounterMessage::Increment(10)).await?;
    system.send(&counter_pid, CounterMessage::Get).await?;
    system.send(&counter_pid, CounterMessage::Reset).await?;
    system.send(&counter_pid, CounterMessage::Get).await?;
    
    // Give some time for messages to be processed
    tokio::time::sleep(Duration::from_millis(100)).await;
    
    // In a real persistent actor, we would:
    // 1. Stop the actor
    // 2. Create a new actor with the same ID
    // 3. The actor would recover its state from stored events and snapshots
    // 4. Continue processing from where it left off
    
    println!("Shutting down system");
    
    Ok(())
} 